package cc.magicjson.example.demo.concurrent;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 并行处理工具类，目前适用场景：entity --> vo
 * author: GengShy
 */

public class ParallelProcessUtil {

    public static void main(String[] args) {
        // create object of LinkedList
        LinkedList<Integer> list
                = new LinkedList<Integer>();

        // Add numbers to end of LinkedList
        list.add(7855642);
        list.add(35658786);
        list.add(5278367);
        list.add(74381793);
        Arrays.stream(list.toArray(new Integer[list.size()])).forEach(System.out::println);

    }


    static final class ThreadPerTaskExecutor implements Executor {
        public void execute(Runnable r) {
            new Thread(r).start();
        }
    }

    private static final boolean useCommonPool =
            (ForkJoinPool.getCommonPoolParallelism() > 1);
    private static final Executor asyncPool = useCommonPool ?
            ForkJoinPool.commonPool() : new ParallelProcessUtil.ThreadPerTaskExecutor();

    public static <T, R> List<List<R>> simpleParallelProcessOfCollection(Collection<T> collection
            , Function<T, R> convertFunction) {

        return parallelProcessOfCollection(Runtime.getRuntime().availableProcessors()
                , collection
                , asyncPool
                , convertFunction);
    }

    public static <T, R> List<List<R>> parallelProcessOfCollection(Collection<T> collection
            , Executor executor
            , Function<T, R> convertFunction) {

        return parallelProcessOfCollection(Runtime.getRuntime().availableProcessors()
                , collection,
                executor,
                convertFunction);
    }

    /**
     * @param threadNum       使用的线程数量
     * @param collection      需要处理的原始集合
     * @param executor        执行器
     * @param convertFunction 结果转换函数
     * @param <T>             原始集合元素类型
     * @param <R>             转换之后的集合元素类型
     * @return 并行处理之后的集合
     * @throws IllegalArgumentException 线程数不合法时
     * @throws NullPointerException     集合或者执行器为空时
     */
    public static <T, R> List<List<R>> parallelProcessOfCollection(int threadNum,
                                                                   Collection<T> collection, Executor executor, Function<T, R> convertFunction) {

        Objects.requireNonNull(collection);
        Objects.requireNonNull(executor);
        if (threadNum <= 0) throw new IllegalArgumentException("线程数不能小于等于0");
        if (threadNum > collection.size()) throw new IllegalArgumentException("线程数不能大于集合大小");
        List<List<T>> lists = splitCollection(collection, threadNum);
        LinkedList<CompletableFuture<List<R>>> completableFutures = new LinkedList<>();
        for (List<T> list : lists) {
            CompletableFuture<List<R>> completableFuture = CompletableFuture.supplyAsync(() ->
                            list.stream().map(convertFunction).collect(Collectors.toList())
                    , executor);
            completableFutures.add(completableFuture);
        }
        int size = completableFutures.size();
        CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[size])).join();
        return completableFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());
    }

    /**
     * 二维集合扁平化
     *
     * @param collections 二维集合
     * @param <T>         集合元素类型
     * @return 转换之后的一维集合
     */
    public static <T> List<T> flattenTwoDimensionalList(Collection<?
            extends Collection<? extends T>> collections) {
        return collections.stream().flatMap(Collection::stream).collect(Collectors.toList());
    }

    private static <T> List<List<T>> splitCollection(Collection<T> collection, int splitNum) {
        int collectionSize = collection.size();
        int splitSize = collectionSize / splitNum;
        int remainder = collectionSize % splitNum;
        ArrayList<List<T>> results = new ArrayList<>();
        int startIndex = 0;
        int endIndex = splitSize;
        for (int i = 0; i < splitNum; i++) {
            results.add(doSplitCollection(collection, startIndex, endIndex));
            startIndex = startIndex + splitSize;
            endIndex = endIndex + splitSize;
        }
        if (remainder != 0) {
            List<T> ts = doSplitCollection(collection, collection.size() - remainder, collection.size());
            results.add(ts);
        }
        return results;
    }


    private static <T> List<T> doSplitCollection(Collection<T> collection,
                                                 int startIndex, int endIndex) {
        if (startIndex < 0 || startIndex >= endIndex || endIndex > collection.size()) {
            throw new IllegalArgumentException("startIndex or endIndex is illegal");
        }
        List<T> subList = new ArrayList<>(endIndex - startIndex);
        Iterator<T> iterator = collection.iterator();
        int index = 0;
        while (iterator.hasNext() && index < endIndex) {
            if (index >= startIndex) {
                subList.add(iterator.next());
            } else {
                iterator.next();
            }
            index++;
        }
        return subList;
    }


}
